#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
NetGearUDP —— UDP 传输 + RTCP 反馈 + Tooth slow-module（按论文）在线推理
----------------------------------------------------------------------
对外仅暴露 NetGearUDP；模块内无全局变量/函数，所有状态内聚在类内。

本版本在原始基础上增强：
1) NEW: 发送端 pacing（独立线程/队列），目标码率 A 的节奏化发送；
2) NEW: 接收端 REMB(Ar) 反馈；发送端融合 A=min(Ar,As) 并做 EMA 平滑；
3) NEW: 帧级 GCC 头（FIRST/LAST + Ti）→ 正确的延迟梯度检测；
4) NEW: 精确 500ms 接收吞吐 Rr 统计；
5) NEW: 发送端 As（loss/编码器约束）接口与基于 lr 的快速止损；
6) CHANGED: 健壮性修复（类型注解、关闭线程、异常保护）。
7) NEW: deliver_pure_payload 开关（默认 True），配合 recv()/recv_payload() 向上层仅交付“纯 payload”，
        满足“用法 A”（应用层无需剥头；直接对 MAGIC/VER 等进行校验）。

新增（本次改动）：
- 媒体优先、冗余退避：新增 _pacer_fec_queue；入队自动识别冗余（FEC/XOR）并分流；发送优先媒体。
- 30ms 短时突发窗口：每次 enqueue 后，为本批数据开 30ms 窗口，窗口内直接发送，不受预算阻挡（仍统一计费）。
"""
import random
import socket
import threading
import time
import logging as log
import struct
import statistics
from collections import deque
from typing import Any, Dict, Optional, Tuple, Union, List, Deque
import os
import json
import base64
from datetime import datetime
import math
import queue
import importlib.util as _imp
import types as _types
from dataclasses import dataclass, field

# ====== 慢模块加载（保持原逻辑，并支持动态导入） ======
# try:
#     # 若你的工程是包方式导入，可在同目录放 tooth_slow_module.py
#     from .tooth_slow_module import ToothSlowModule, load_weights  # type: ignore
#     _SLOW_AVAILABLE = True
# except Exception:
#     ToothSlowModule = None  # type: ignore
#     load_weights = None     # type: ignore
#     _SLOW_AVAILABLE = False


# def _load_slow_module_dynamic(path: str):
#     """NEW: 从给定文件路径动态加载 tooth_slow_module.py（或 slow_module.py）。"""
#     spec = _imp.spec_from_file_location("tooth_slow_module_dyn", path)
#     if spec is None or spec.loader is None:
#         raise ImportError(f"spec_from_file_location failed for {path}")
#     mod = _imp.module_from_spec(spec)  # type: _types.ModuleType
#     spec.loader.exec_module(mod)       # type: ignore[attr-defined]
#     return mod


# ===================== GCC 相关结构（发送端/接收端共享） =====================
@dataclass
class _GCCState:
    """
    NEW: GCC 相关状态聚合
    字段含义：
      enabled: 是否启用 GCC 模式；
      as_bps:  发送端 loss/编码器侧上限（As），默认 +inf；
      ar_bps:  接收端 REMB 推荐码率（Ar），默认 +inf；
      target_bps: 融合目标码率 A=min(Ar,As)，pacer 按其节奏发包；

      pacing_factor: 当队列拥塞时的瞬时加速倍数（>1 表示加速）；
      pacer_sleep_ns: pacer 内部每个包发送后的最小休眠（纳秒）；

      m_t:   延迟梯度估计；
      gamma: 自适应阈值（|m| 的平滑上界，用于过载判断）；
      ku/kd: γ 的上/下跟随速率（GCC 经典：ku=0.01, kd=0.00018）；
      Q:     近似 Kalman 的过程噪声权重；
      beta_noise: 残差方差 EWMA 的平滑系数；
      resid_var: 残差方差估计；
      state:  FSM 当前态（Increase/Hold/Decrease）；

      prev_Ti/prev_ti: 前一帧的“发送首包时间/最后一包到达时间”；
      cur_*:  当前帧跟踪（frame_id、首/末包到达等）；

      recv_bytes_500ms: 最近 500ms 到达的 DATA 字节数队列（(ts, bytes)）；
      recv_bytes_acc:   累计字节（辅助合并统计）；
      recv_win_start:   吞吐统计窗口起点；

      eta_inc:   Increase 慢加速倍数；
      alpha_dec: Decrease 快减速系数（乘 Rr）；
      rr_ub_factor: REMB 输出不超过 1.5×Rr 的护栏；
    """
    enabled: bool = False

    # 发送端：loss-based 控制器（As） & 融合后目标 A
    as_bps: float = float("inf")
    ar_bps: float = float("inf")
    target_bps: float = float("inf")  # A = min(Ar, As)
    target_bps_ema: float = float("inf")  # NEW: A 的 EMA 平滑值

    # 发送端：pacer
    pacing_factor: float = 1.5
    pacer_sleep_ns: int = 200_000  # 200us
    backlog_boost_threshold: int = 256  # NEW: 队列积压触发加速的阈值（包数）

    # 接收端：到达时间滤波/阈值/FSM
    m_t: float = 0.0
    gamma: float = 0.0
    ku: float = 0.05
    kd: float = 1e-5
    Q: float = 5e-3
    beta_noise: float = 0.98
    resid_var: float = 0.0
    state: str = "Hold"

    # 接收端：上一帧 (Ti, ti)
    prev_Ti: Optional[float] = None
    prev_ti: Optional[float] = None

    # 接收端：当前帧跟踪
    cur_frame_id: Optional[int] = None
    cur_Ti: Optional[float] = None
    cur_first_seen: Optional[float] = None
    cur_last_seen: Optional[float] = None
    cur_received_any: bool = False

    # 接收端：吞吐测量（500ms 窗）
    recv_bytes_500ms: Deque[Tuple[float, int]] = field(default_factory=lambda: deque(maxlen=2048))
    recv_bytes_acc: int = 0
    recv_win_start: float = time.time()

    # 工况参数
    eta_inc: float = 1.05
    alpha_dec: float = 0.85
    rr_ub_factor: float = 1.5


class NetGearUDP:
    """
    角色：
      - receive_mode=False: 发送端（发 DATA/PING；收 RTCP/PONG/ACK）
      - receive_mode=True : 接收端（收 DATA/PING；发 RTCP/PONG）

    关键 API（对外）：
      - send(data: bytes, pkt_type: int=None) -> None
      - enqueue_frame_pkts(pkts: List[bytes], frame_id: int) -> None
      - set_target_bps(bps: float) -> None
      - set_as_bps(bps: float) -> None
      - enable_gcc(enable: bool=True) -> None
      - get_gcc_target_bps() -> float
      - recv() / recv_payload()  # NEW: 直接返回纯 payload
      - get_rtcp_report() / get_rtt_stats() / get_stats() / reset_stats() / close()
      - get_slow_prediction() / get_network_status()
      - udp_ack_packet(packet_id: int, ts_ms: Optional[int]=None) -> None

    数据包类型：
      DATA/RES/SV_DATA/ACK_PACKET/PING/PONG/RTCP_REPORT/REMB
    """

    DEFAULT_MTU = 1500
    RECV_QUEUE_MAXLEN = 32768

    PKT_DATA = 0
    PKT_RES = 1
    PKT_SV_DATA = 2
    PKT_TERM = 3
    PKT_ACK_FRAME = 4
    PKT_ACK_PACKET = 5
    PKT_PING = 6
    PKT_PONG = 7
    PKT_RTCP_REPORT = 8
    PKT_REMB = 9               # NEW: REMB (Ar) 报文
    PKT_RTCP_REQ = 10   # NEW: 发送端→接收端：请求在下一个 RTCP_REPORT 中回显时间戳

    # —— GCC data 子头 ——（仅在 GCC 模式下由发送端自动加/接收端自动解析）
    # body 结构: b'GC'(2B) + flags(1B) + frame_id(u32) + Ti(double) + payload...
    # flags: bit0=FIRST, bit1=LAST
    GCC_MAGIC = b'GC'
    GCC_HDR_FMT = '>2sBId'
    GCC_HDR_LEN = struct.calcsize(GCC_HDR_FMT)
    GCC_FLAG_FIRST = 0x01
    GCC_FLAG_LAST = 0x02

    _LOGGER = log.getLogger("NetGearUDP")
    _LOGGER.setLevel(log.DEBUG)
    if not _LOGGER.handlers:
        _h = log.StreamHandler()
        _f = log.Formatter("[%(levelname)s] %(asctime)s %(name)s: %(message)s")
        _h.setFormatter(_f)
        _LOGGER.addHandler(_h)

    def __init__(
        self,
        address: str = "0.0.0.0",
        port: Union[int, str] = 5556,
        protocol: str = "udp",
        receive_mode: bool = False,
        logging: bool = True,
        # NEW: 是否向上层仅交付“纯 payload”（剥 type/seq 及 GCC 子头）；默认 True，契合“用法 A”
        deliver_pure_payload: bool = True,
        **options
    ) -> None:
        self._logging = bool(logging)
        self._addr = address
        self._port = int(port)
        if protocol.lower() != "udp":
            raise ValueError("NetGearUDP 仅支持 protocol='udp'。")

        self._mtu = int(options.get("mtu", self.DEFAULT_MTU))
        self._recv_buf_size = int(options.get("recv_buffer_size", 16 * 1024 * 1024))
        self._send_buf_size = int(options.get("send_buffer_size", 16 * 1024 * 1024))
        self._queue_maxlen = int(options.get("queue_maxlen", self.RECV_QUEUE_MAXLEN))
        self._enable_seq = bool(options.get("enable_seq_header", True))
        self._rtcp_interval_ms = int(options.get("rtcp_interval_ms", 100))

        # NEW: 交付纯 payload 的开关（True→交付“纯 payload”；False→交付“含 GCC 子头的 body”）
        self._deliver_pure_payload: bool = bool(deliver_pure_payload)

        # ---- slow-module 配置（保持原逻辑） ----
        self._slow_log_dir: str = str(options.get("slow_log_dir", 
                                                  "/data/data/com.termux/files/usr/var/lib/proot-distro/installed-rootfs/ubuntu/root/slow_logs_sender" 
                                                  if not receive_mode else "/data/shb/viduce/slow_logs_receiver"))
        self._slow_weights_path: str = str(options.get("slow_weights_path", "./tmp/slow_module_weights.json"))
        self._slow_hist_n: int = int(options.get("slow_hist_n", 10))

        self._slow_slots_per_window: int = int(options.get("slow_slots_per_window", 32))
        self._slow_bytes_per_window: int = (self._slow_slots_per_window + 7) // 8
        self._slow_module_path: Optional[str] = options.get("slow_module_path")

        # 创建日志目录
        try:
            os.makedirs(self._slow_log_dir, exist_ok=True)
        except Exception:
            pass
        ts_name = datetime.now().strftime("%Y%m%d_%H%M%S")
        self._slow_jsonl_path = os.path.join(self._slow_log_dir, f"session_{ts_name}.jsonl")

        # ---- 发送/接收统计 ----
        self._sent_packets = 0
        self._total_packets_received = 0

        # RTT（发送端 PING/PONG 估计，单位 ms）
        self._rtt_samples = deque(maxlen=512)
        self._rtt_ema_ms: Optional[float] = None
        self._last_rtcp_req_ts: Optional[float] = None  # NEW: 最近一次 RTCP_REQ 的发送时刻

        # RTCP（仅发送端使用拉取队列）
        self._rtcp_reports: deque = deque(maxlen=1024)
        self._last_rtcp: Optional[Dict[str, Any]] = None

        # 接收端窗口统计（用于 lr/la 与接收端 P_ls）
        self._win_start_ts = time.time()
        self._win_rx_count = 0
        self._win_expected_count = 0
        self._last_seq_rx: Optional[int] = None  # FIXED: Optional[int]
        self._rx_missing_seq_times: List[float] = []

        # ====== 发送端：慢模块输入历史 ======
        self._hist_ls_bits: deque = deque(maxlen=self._slow_hist_n)
        self._hist_lr: deque = deque(maxlen=self._slow_hist_n)
        self._hist_la: deque = deque(maxlen=self._slow_hist_n)

        # ====== 发送端：包级 ACK 记录（原逻辑） ======
        self._acked_seq: set = set()
        self._tx_records: deque = deque()
        self._tx_index: Dict[int, Dict[str, Any]] = {}
        self._tx_pkt_log_path = os.path.join(self._slow_log_dir, f"tx_packets_sender_{ts_name}.jsonl")
        self._last_tx_dump_ts = time.time()

        # ====== 慢模块 ======
        self._slow_pred: Dict[str, float] = {"lr_f": 0.0, "la_f": 0.0}

        # # 动态加载 slow module（可选）
        # global _SLOW_AVAILABLE, ToothSlowModule, load_weights
        # if (not _SLOW_AVAILABLE) and self._slow_module_path:
        #     try:
        #         _mod = _load_slow_module_dynamic(self._slow_module_path)
        #         ToothSlowModule = getattr(_mod, "ToothSlowModule")
        #         load_weights = getattr(_mod, "load_weights", None)
        #         _SLOW_AVAILABLE = True
        #         if self._logging:
        #             self._LOGGER.info(f"[SlowModule] Loaded from file: {self._slow_module_path}")
        #     except Exception as e:
        #         if self._logging:
        #             self._LOGGER.warning(f"[SlowModule] dynamic import failed: {e}")
        #         _SLOW_AVAILABLE = False

        # if _SLOW_AVAILABLE:
        #     try:
        #         self._slow_model = ToothSlowModule()  # type: ignore
        #         if os.path.isfile(self._slow_weights_path) and (load_weights is not None):
        #             load_weights(self._slow_model, self._slow_weights_path)  # type: ignore
        #             if self._logging:
        #                 self._LOGGER.info(f"[SlowModule] Loaded weights: {self._slow_weights_path}")
        #     except Exception as e:
        #         self._slow_model = None  # type: ignore
        #         if self._logging:
        #             self._LOGGER.warning(f"[SlowModule] init failed: {e}")
        # else:
        #     self._slow_model = None  # type: ignore
        #     if self._logging:
        #         self._LOGGER.warning("[SlowModule] slow_module.py 未可用，在线推理被禁用。")

        # ---- UDP 套接字 ----
        self._recv_mode = bool(receive_mode)
        self._queue: deque = deque(maxlen=self._queue_maxlen)
        self._peer_addr: Optional[Tuple[str, int]] = None

        self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, self._recv_buf_size)
            self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, self._send_buf_size)
            self._sock.setblocking(False)
        except Exception as e:
            if self._logging:
                self._LOGGER.warning(f"Socket option setting failed: {e}")

        self._terminate = False

        if self._recv_mode:
            self._sock.bind((self._addr, self._port))
            if self._logging:
                self._LOGGER.info(f"[UDP] Bind at {self._addr}:{self._port}")
        else:
            self._peer_addr = (self._addr, self._port)
            if self._logging:
                self._LOGGER.info(f"[UDP] Send to {self._addr}:{self._port}")
        local_bind_port = options.get("local_bind_port", None)
        if (not self._recv_mode) and local_bind_port:
            self._sock.bind(("0.0.0.0", int(local_bind_port)))

        # ---- 异步日志线程（保持原逻辑） ----
        self._logq: "queue.SimpleQueue[Tuple[str,str]]" = queue.SimpleQueue()
        self._log_thr = threading.Thread(target=self._log_sink_loop, name="UDP-LOG", daemon=True)
        self._log_thr.start()

        # ---- I/O 与 RTCP 线程 ----
        self._rx_thread = threading.Thread(target=self._io_loop, name="UDP-RX", daemon=True)
        self._rx_thread.start()

        self._rtcp_thread = threading.Thread(target=self._rtcp_loop, name="UDP-RTCP", daemon=True)
        self._rtcp_thread.start()

        # ====== NEW: GCC/pacer 状态与线程 ======
        self._gcc = _GCCState(enabled=False)
        self._A_ema_alpha = float(options.get("gcc_A_ema_alpha", 0.3))  # NEW: A 的 EMA 平滑系数
        self._loss_lr_thresh = float(options.get("loss_lr_thresh", 0.05))  # NEW: lr 超过此阈值触发 As 止损
        self._loss_as_decay = float(options.get("loss_as_decay", 0.6))     # NEW: 止损时 As *= 0.6
        self._pacer_lock = threading.Lock()

        # 媒体主队列（DATA）
        self._pacer_queue: deque = deque()  # 元素：(payload, pkt_type, frame_id, deadline)
        self._pacer_thread = threading.Thread(target=self._pacer_loop, name="UDP-PACER", daemon=True)
        self._pacer_max_pkts = int(options.get("pacer_max_pkts", 8192))  # 队列硬上限（包）

        # === NEW: FEC/XOR 冗余专用队列与容量、统计 ===
        self._pacer_fec_queue: deque = deque()  # 元素同上
        self._pacer_fec_max_pkts = int(options.get("pacer_fec_max_pkts", self._pacer_max_pkts))
        self._pacer_fec_drop_overflow = 0
        self._pacer_fec_drop_expired = 0

        self._frame_ttl_ms   = float(options.get("frame_ttl_ms", 300))   # 单帧过期阈值（毫秒）
        self._pacer_drop_expired = int(0)     # 统计：过期丢弃的包数（媒体）
        self._pacer_drop_overflow = int(0)    # 统计：溢出丢弃的包数（媒体）

        # === NEW: 30ms 短时突发窗口（按入队批次发完；窗口内不受 bytes_budget 限制） ===
        self._burst_until_ts: float = 0.0     # 当前突发窗口的截止时间戳
        self._burst_bytes_left: int = 0       # 窗口内允许“借用”的剩余字节；窗口内不看预算，但仍统一计费

        self._pacer_thread.start()

    # ===================== NEW: 外部控制接口 =====================
    def enable_gcc(self, enable: bool = True) -> None:
        """开启/关闭 GCC；关闭时 target_bps 重置为 +inf，行为回到非 GCC 的 send() 直发。"""
        self._gcc.enabled = bool(enable)
        if not enable:
            self._gcc.ar_bps = float("inf")
            self._gcc.as_bps = float("inf")
            self._gcc.target_bps = float("inf")
            self._gcc.target_bps_ema = float("inf")

    def set_target_bps(self, bps: float) -> None:
        """上层（含编码器/预算）设置/覆盖 A 的期望值（bps）。"""
        b = float(bps)
        if b <= 0:
            b = 1.0
        self._gcc.target_bps = float(b)
        self._apply_A_ema()

    def set_as_bps(self, bps: float) -> None:
        """NEW: 设置发送端 loss/编码器侧约束 As（bps）。"""
        b = float(bps)
        if b <= 0:
            b = 1.0
        self._gcc.as_bps = float(b)
        # 重新融合并平滑
        self._fuse_A_and_apply_ema()

    def get_gcc_target_bps(self) -> float:
        """返回当前融合后的目标 A（bps）。若未启用或未知，返回 +inf。"""
        return float(self._gcc.target_bps_ema if math.isfinite(self._gcc.target_bps_ema) else self._gcc.target_bps)
        # return self._gcc.target_bps

    # --- 工具：从原始 raw 解析外层统一包头，判断是否为 FEC/XOR ---
    def _is_fec_like_packet(self, raw: bytes) -> bool:
        """
        返回 True 表示这是冗余包（FEC/XOR），False 表示媒体 DATA。
        统一外层头：MAGIC(2)='SP', VER(1)=1, TYPE(1) in {1,2,3}；TYPE: 1=DATA, 2=FEC, 3=XOR。
        """
        try:
            if raw is None or len(raw) < 4:
                return False
            if raw[0:2] != b"SP":
                return False
            typ = raw[3]
            return typ in (2, 3)
        except Exception:
            return False

    def _maybe_drop_overflow(self, total_add: int) -> None:
        with self._pacer_lock:
            while (len(self._pacer_queue) + len(self._pacer_fec_queue) + total_add) > (self._pacer_max_pkts + self._pacer_fec_max_pkts):
                # 优先丢冗余队列
                if self._pacer_fec_queue:
                    self._pacer_fec_queue.popleft()
                    self._pacer_fec_drop_overflow += 1
                elif self._pacer_queue:
                    self._pacer_queue.popleft()
                    self._pacer_drop_overflow += 1
                else:
                    break

    def enqueue_frame_pkts(self, pkts: List[bytes], frame_id: int) -> None:
        """
        NEW: 将一帧的多个分片放入 pacer 队列（GCC 模式）。
        - 自动加 GCC 轻量 data 头（含首包发送时间 Ti 与 FIRST/LAST 标志）
        - 媒体/冗余分流：媒体(DATA)→_pacer_queue；FEC/XOR(RES)→_pacer_fec_queue
        - 每次入队后开启 30ms 窗口：在窗口内本批数据直接发完，不受 bytes_budget 阻挡（仍统一计费）
        - 非 GCC 模式下，兜底：直接 send()
        """
        if not self._gcc.enabled:
            for raw in pkts:
                self.send(raw)
            return

        Ti = time.time()
        n = len(pkts)
        deadline = Ti + (self._frame_ttl_ms / 1000.0)

        # 预检查容量（优先丢冗余）
        self._maybe_drop_overflow(n)

        batch_bytes = 0  # 本次入队的总 payload（含 GCC 头）的估算，用于突发窗口额度

        with self._pacer_lock:
            for idx, raw in enumerate(pkts):
                # is_fec = self._is_fec_like_packet(raw)

                # FIRST/LAST 标志
                flags = 0
                if idx == 0:
                    flags |= self.GCC_FLAG_FIRST
                if idx == n - 1:
                    flags |= self.GCC_FLAG_LAST

                # 轻量 GCC 头 + 原始分片
                hdr = struct.pack(self.GCC_HDR_FMT, self.GCC_MAGIC, flags & 0xFF, int(frame_id), float(Ti))
                payload = hdr + raw

                # 统计本批字节（用于 30ms 窗口）
                batch_bytes += len(payload)

                # TODO: 入队（媒体优先；冗余进专队列）
                # if is_fec:
                #     if len(self._pacer_fec_queue) >= self._pacer_fec_max_pkts and self._pacer_fec_queue:
                #         self._pacer_fec_queue.popleft()
                #         self._pacer_fec_drop_overflow += 1
                #     # 注意：冗余标记为 PKT_RES（之前误写为 PKT_DATA，现已修正）
                #     self._pacer_fec_queue.append((payload, self.PKT_RES, int(frame_id), float(deadline)))
                # else:
                if len(self._pacer_queue) >= self._pacer_max_pkts and self._pacer_queue:
                    self._pacer_queue.popleft()
                    self._pacer_drop_overflow += 1
                self._pacer_queue.append((payload, self.PKT_DATA, int(frame_id), float(deadline)))

            # === NEW: 开启/合并 30ms 突发窗口 ===
            now2 = time.time()
            until = now2 + 0.030  # 30ms
            # 若已有窗口，则把截止时间向后延，并叠加额度
            if until > self._burst_until_ts:
                self._burst_until_ts = float(until)
            self._burst_bytes_left += int(batch_bytes)

    # ===================== 工具 =====================
    @staticmethod
    def _now_ms() -> int:
        return int(time.time() * 1000)

    # ===================== 类型判断（供上层使用） =====================
    def is_ack_frame(self, pkt_type: int) -> bool:
        return int(pkt_type) == self.PKT_ACK_FRAME

    def is_ack_packet(self, pkt_type: int) -> bool:
        return int(pkt_type) == self.PKT_ACK_PACKET

    def is_res(self, pkt_type: int) -> bool:
        return int(pkt_type) == self.PKT_RES

    # ===================== 包级 ACK 入口（发送端诊断） =====================
    def udp_ack_packet(self, packet_id: int, ts_ms: Optional[int] = None) -> None:
        try:
            self._acked_seq.add(int(packet_id))
        except Exception:
            pass
        try:
            rec = self._tx_index.get(int(packet_id))
            if rec is not None and rec.get("ts_ack") is None:
                rec["ts_ack"] = float(time.time()) if ts_ms is None else (float(ts_ms)/1000.0)
        except Exception:
            pass

    # ===================== 慢模块预测/网络状态查询（原逻辑 + GCC 指标） =====================
    def get_slow_prediction(self) -> Dict[str, float]:
        return dict(self._slow_pred)

    def get_network_status(self) -> Dict[str, Any]:
        return {
            "rtcp": (None if self._last_rtcp is None else dict(self._last_rtcp)),
            "rtt": self.get_rtt_stats(),
            "slow_pred": self.get_slow_prediction(),
            "hist": {
                "lr": list(self._hist_lr),
                "la": list(self._hist_la),
                "ls_windows": len(self._hist_ls_bits),
            },
            "gcc": {
                "enabled": self._gcc.enabled,
                "A_bps": self.get_gcc_target_bps(),
                "Ar_bps": self._gcc.ar_bps,
                "As_bps": self._gcc.as_bps,
                "A_raw": self._gcc.target_bps,
                "A_ema": self._gcc.target_bps_ema,
                "m_t": self._gcc.m_t,
                "gamma": self._gcc.gamma,
                "state": self._gcc.state,
                "pacer_queue": len(self._pacer_queue),
            },
            # NEW: FEC 队列可观测
            "pacer": {
                "expired": int(self._pacer_drop_expired),
                "overflow": int(self._pacer_drop_overflow),
                "queue_len": int(len(self._pacer_queue)),
                "queue_cap": int(self._pacer_max_pkts),
                "fec_expired": int(self._pacer_fec_drop_expired),
                "fec_overflow": int(self._pacer_fec_drop_overflow),
                "fec_queue_len": int(len(self._pacer_fec_queue)),
                "fec_queue_cap": int(self._pacer_fec_max_pkts),
                "burst_left": int(self._burst_bytes_left),
                "burst_until_ms": int(max(0.0, (self._burst_until_ts - time.time()) * 1000.0)),
            }
        }

    # ===================== 发送（保持原逻辑） =====================
    def send(self, frame: Union[bytes, bytearray, memoryview], pkt_type: int = None) -> None:
        """
        发送一个逻辑包；若启用 seq 头，则对 DATA/RES/SV_DATA/ACK_PACKET 写入 4B 序号。
        GCC 模式下，通常不直接调用本函数发 DATA，而是通过 enqueue_frame_pkts() 进 pacer。
        """
        if frame is None:
            return
        if pkt_type is None:
            pkt_type = self.PKT_DATA
        if self._peer_addr is None and not self._recv_mode:
            raise RuntimeError("No peer address configured for sender.")

        custom_hdr_len = 1 + (4 if (self._enable_seq and pkt_type in
                                    (self.PKT_DATA, self.PKT_RES, self.PKT_SV_DATA, self.PKT_ACK_PACKET)) else 0)
        max_payload = self._mtu - 28 - custom_hdr_len
        if max_payload <= 0:
            raise ValueError(f"MTU 设置过小：mtu={self._mtu}。")
        data = bytes(frame)
        if len(data) > max_payload:
            raise ValueError(f"数据过大：len(data)={len(data)} > {max_payload}。")

        out = bytearray()
        out.append(int(pkt_type) & 0xFF)

        seq_for_log: Optional[int] = None
        if self._enable_seq and pkt_type in (self.PKT_DATA, self.PKT_RES, self.PKT_SV_DATA, self.PKT_ACK_PACKET):
            if not hasattr(self, "_seq_tx"):
                self._seq_tx = 0  # type: ignore[attr-defined]
            self._seq_tx = int((self._seq_tx + 1) & 0xFFFFFFFF)  # type: ignore[attr-defined]
            out.extend(struct.pack(">I", self._seq_tx))  # type: ignore[attr-defined]
            seq_for_log = int(self._seq_tx)  # type: ignore[attr-defined]

        out.extend(data)
        try:
            if self._peer_addr:
                try:
                    # TODO: 1%的概率丢包
                    if random.random() > 0.01:
                        self._sock.sendto(out, self._peer_addr)
                    self._sent_packets += 1
                except BlockingIOError as e:
                    if e.errno == 11:  # Resource temporarily unavailable
                        # 当作丢包，不 raise
                        if self._logging:
                            self._LOGGER.warning("send buffer full -> treat as drop")
                    else:
                        raise

            # 发送端滚动记录DATA包（仅发送端 && DATA）
            if (not self._recv_mode) and (pkt_type == self.PKT_DATA) and (seq_for_log is not None):
                now = time.time()
                rec = {"seq": int(seq_for_log), "ts_send": float(now), "ts_ack": None}
                self._tx_records.append(rec)
                self._tx_index[int(seq_for_log)] = rec
                if len(self._tx_records) > 200000:
                    self._dump_old_tx_records(force=True)

        except Exception as e:
            if self._logging:
                self._LOGGER.warning(f"send error: {e}")
            raise

    # ===================== 接收 API（新增 recv_payload） =====================
    def recv(self) -> Optional[Dict[str, Any]]:
        """
        从内部队列取出一条消息：
          返回 dict: {"pkt_type": int, "data": bytes, "seq": Optional[int]}
        注意：若 deliver_pure_payload=True（默认），接收端 DATA/RES/SV_DATA 的 "data"
              已经是“纯 payload”（库已剥离 type/seq 及 GCC 子头），可直接做 MAGIC/VER 校验。
        """
        if self._queue:
            return self._queue.popleft()
        return None

    def recv_payload(self) -> Optional[bytes]:
        """
        NEW: 直接返回“纯 payload”。
        - 若队列中消息为 DATA/RES/SV_DATA，返回其 data（已是纯 payload）
        - 对于控制类包，返回 None
        """
        msg = self.recv()
        if not msg:
            return None
        pkt_type = int(msg.get("pkt_type", -1))
        if pkt_type in (self.PKT_DATA, self.PKT_RES, self.PKT_SV_DATA):
            return msg.get("data", b"")
        return None

    def get_rtcp_report(self) -> Optional[Dict[str, Any]]:
        if self._rtcp_reports:
            return self._rtcp_reports.popleft()
        return None

    def get_rtt_stats(self) -> Dict[str, Any]:
        median_ms = statistics.median(self._rtt_samples) if self._rtt_samples else None
        return {
            "ema_ms": None if self._rtt_ema_ms is None else float(self._rtt_ema_ms),
            "median_ms": None if median_ms is None else float(median_ms),
            "samples": int(len(self._rtt_samples)),
        }

    def get_stats(self) -> Dict[str, int]:
        return {"total_packets_received": int(self._total_packets_received),
                "sent_packets": int(self._sent_packets)}

    def reset_stats(self) -> None:
        self._win_start_ts = time.time()
        self._win_rx_count = 0
        self._win_expected_count = 0
        self._last_seq_rx = None
        self._rx_missing_seq_times.clear()
        self._total_packets_received = 0
        self._sent_packets = 0

    def close(self) -> None:
        self._terminate = True
        try:
            if hasattr(self, "_rtcp_thread") and self._rtcp_thread.is_alive():
                self._rtcp_thread.join(timeout=1.0)
        except Exception:
            pass
        try:
            if hasattr(self, "_rx_thread") and self._rx_thread.is_alive():
                self._rx_thread.join(timeout=1.0)
        except Exception:
            pass
        try:
            if hasattr(self, "_log_thr") and self._log_thr.is_alive():
                time.sleep(0.1)
        except Exception:
            pass
        try:
            if hasattr(self, "_pacer_thread") and self._pacer_thread.is_alive():
                self._pacer_thread.join(timeout=1.0)  # NEW: 等待 pacer 线程退出
        except Exception:
            pass
        try:
            self._sock.close()
        except Exception:
            pass

    # ===================== I/O 线程（收包/解析） =====================
    def _io_loop(self) -> None:
        import select
        while not self._terminate:
            try:
                ready = select.select([self._sock], [], [], 0.05)
                if not ready[0]:
                    # 空闲：发送端滚动落盘
                    if not self._recv_mode:
                        self._dump_old_tx_records()
                    continue

                pkt, peer = self._sock.recvfrom(65535)
                now_r = time.time()
                if not pkt:
                    continue
                self._peer_addr = peer

                pkt_type = pkt[0]
                offset = 1
                seq: Optional[int] = None

                # 可选序号头
                if self._enable_seq and pkt_type in (self.PKT_DATA, self.PKT_RES, self.PKT_SV_DATA, self.PKT_ACK_PACKET):
                    if len(pkt) < 1 + 4:
                        if self._logging:
                            self._LOGGER.warning("packet too short for seq header")
                        continue
                    seq = struct.unpack(">I", pkt[offset:offset+4])[0]
                    offset += 4

                body = pkt[offset:]

                # === 控制面（不上抛） ===
                if pkt_type == self.PKT_ACK_PACKET:
                    self._total_packets_received += 1
                    try:
                        if len(body) >= 4:
                            data_seq = struct.unpack(">I", body[:4])[0]
                            self.udp_ack_packet(int(data_seq))
                    except Exception:
                        pass
                    continue

                if pkt_type == self.PKT_RTCP_REQ and self._recv_mode:
                    if len(body) >= 8:
                        try:
                            (ts_req,) = struct.unpack(">d", body[:8])
                            self._last_rtcp_req_ts = float(ts_req)  # NEW: 缓存最后一次请求时间
                        except Exception:
                            pass
                    continue

                if pkt_type == self.PKT_RTCP_REPORT and not self._recv_mode:
                    # RTCP 报告：lr/la/位图... + （NEW）回显RTT时间戳
                    if len(body) >= 28:
                        try:
                            # 头部（28B）
                            win_start_ts, lr, la, rx_pkts, win_ms, last_seq = struct.unpack(">dffffI", body[:28])
                            rep = {
                                "win_start_ts": float(win_start_ts),
                                "lr": float(lr),
                                "la": float(la),
                                "rx_pkts": int(rx_pkts),
                                "win_ms": float(win_ms),
                                "last_seq_rx": int(last_seq),
                            }
                            self._last_rtcp = rep
                            if len(self._rtcp_reports) < self._rtcp_reports.maxlen:
                                self._rtcp_reports.append(rep)

                            # NEW: 根据最新 lr 触发 As 止损（快速抑制过载）
                            self._update_As_from_loss(float(lr))

                            # ----- 解析位图与 RTT 回显 -----
                            bits = None
                            base_off = 28
                            off = base_off

                            # 位图长度 + 位图本体
                            if len(body) > off:
                                blen = body[off]
                                off += 1
                                if len(body) >= off + blen and blen > 0:
                                    bmap = body[off:off + blen]
                                    off += blen
                                    # LSB-first 展开
                                    S = blen * 8
                                    bits = []
                                    for j in range(S):
                                        byte = bmap[j // 8]
                                        bits.append((byte >> (j % 8)) & 1)

                            # NEW: RTT 回显（echo 区）
                            # 结构：echo_flag(1B) [+ echo_ts(double, 8B)]
                            if len(body) > off:
                                echo_flag = body[off]
                                off += 1
                                if echo_flag == 1 and len(body) >= off + 8:
                                    (echo_ts,) = struct.unpack(">d", body[off:off + 8])
                                    # 以“收到 RTCP_REPORT 的当前时刻 - 对端回显的请求发送时刻”作为 RTT
                                    rtt_ms = (now_r - float(echo_ts)) * 1000.0
                                    # 写入 RTT 样本与 EMA（复用既有统计接口）
                                    self._rtt_samples.append(rtt_ms)
                                    if self._rtt_ema_ms is None:
                                        self._rtt_ema_ms = rtt_ms
                                    else:
                                        self._rtt_ema_ms = 0.8 * float(self._rtt_ema_ms) + 0.2 * rtt_ms

                            # 保持你原有的慢模块输入与日志落盘
                            if bits is not None:
                                self._hist_ls_bits.append(bits)
                                self._hist_lr.append(float(lr))
                                self._hist_la.append(float(la))
                                self._write_jsonl_sample(
                                    float(win_start_ts), float(lr), float(la), float(win_ms),
                                    bits_override=bits
                                )
                                # self._run_slow_forward()
                        except Exception as e:
                            if self._logging:
                                self._LOGGER.warning(f"parse RTCP report error: {e}")
                    continue

                if pkt_type == self.PKT_REMB and not self._recv_mode:
                    # NEW: 接收 REMB（Ar）
                    if len(body) >= 8:
                        try:
                            (ar_bps,) = struct.unpack(">d", body[:8])
                            self._gcc.ar_bps = float(max(1.0, ar_bps))
                            self._fuse_A_and_apply_ema()  # 融合并平滑
                        except Exception:
                            pass
                    continue

                # === 数据面：接收端统计 +（可选）GCC arrival 解析 ===
                if self._recv_mode and self._enable_seq and (pkt_type in (self.PKT_DATA, self.PKT_RES, self.PKT_SV_DATA)):
                    now = time.time()
                    self._win_rx_count += 1
                    if seq is not None:
                        if self._last_seq_rx is None:
                            self._last_seq_rx = seq
                            self._win_expected_count += 1
                        else:
                            gap = seq - self._last_seq_rx
                            if gap <= 0:
                                self._win_expected_count += 1
                            else:
                                self._win_expected_count += gap
                                if gap > 1:
                                    self._rx_missing_seq_times.extend([now] * (gap - 1))
                            self._last_seq_rx = seq

                    # NEW: 精确累计接收的 DATA/RES 字节，用于 500ms Rr 估计
                    raw_len = len(body) + offset
                    if self._gcc.enabled:
                        self._gcc.recv_bytes_500ms.append((now, raw_len*10))

                    # ——（关键）GCC 子头剥离（deliver_pure_payload=True 时做）——
                    if self._deliver_pure_payload and len(body) >= self.GCC_HDR_LEN:
                        try:
                            magic, flags, frame_id, Ti = struct.unpack(self.GCC_HDR_FMT, body[:self.GCC_HDR_LEN])
                            if magic == self.GCC_MAGIC:
                                # 推进到达时间状态机
                                self._gcc_got_gcc_data(int(frame_id), float(Ti),
                                                       bool(flags & self.GCC_FLAG_FIRST),
                                                       bool(flags & self.GCC_FLAG_LAST), now)
                                # 交付上层的数据：剥掉 GCC 子头 → 纯 payload
                                body = body[self.GCC_HDR_LEN:]
                        except Exception:
                            pass

                # 入队给上层（仅非控制包）
                if len(self._queue) < self._queue_maxlen:
                    self._queue.append({
                        "pkt_type": int(pkt_type),
                        "data": body,
                        "seq": None if seq is None else int(seq),
                    })

            except socket.error as e:
                err = getattr(e, "errno", None)
                if err in (socket.EAGAIN, socket.EWOULDBLOCK):
                    time.sleep(0.0001)
                    continue
                if not self._terminate and self._logging:
                    self._LOGGER.error(f"recv error: {e}")
            except Exception as e:
                if not self._terminate and self._logging:
                    self._LOGGER.error(f"recv error: {e}")

    # ===================== RTCP/PING 线程（接收端 REMB 发送） =====================
    def _rtcp_loop(self) -> None:
        """
        - 发送端：定期发 PING（这里用 RTCP_REQ 实现），基于回显估计 RTT（毫秒）
        - 接收端：每 rtcp_interval_ms 汇总一次窗口，发送 RTCP_REPORT；
                 若启用 GCC，则同时基于到达时间滤波输出 REMB(Ar)
        """
        ping_interval = max(2 * self._rtcp_interval_ms, 100) / 1000.0
        last_ping_ts = time.time()

        while not self._terminate:
            now = time.time()

            # 发送端：周期性 RTCP_REQ（代替 PING）
            if not self._recv_mode and self._peer_addr and (now - last_ping_ts) >= ping_interval:
                last_ping_ts = now
                try:
                    now = time.time()
                    req_body = struct.pack(">d", now)  # 仅携带发送时刻
                    out = bytes([self.PKT_RTCP_REQ]) + req_body
                    # TODO: 1%的概率丢包
                    if random.random() > 0.01:
                        self._sock.sendto(out, self._peer_addr)  # 直接发→绕过 pacer
                except Exception as e:
                    if self._logging:
                        self._LOGGER.debug(f"send RTCP_REQ error: {e}")

            # 接收端：汇总 RTCP + REMB
            if self._recv_mode and (now - self._win_start_ts) * 1000.0 >= self._rtcp_interval_ms:
                rx = self._win_rx_count
                expected = max(self._win_expected_count, rx)
                lost = max(expected - rx, 0)
                lr = (lost / expected) if expected > 0 else 0.0

                la = 0.0
                n = len(self._rx_missing_seq_times)
                if n > 1:
                    centroid = sum(self._rx_missing_seq_times) / n
                    denom = sum(abs(t - centroid) for t in self._rx_missing_seq_times) + 0.5
                    la = float(n) / denom if denom > 0 else float(n)

                win_ms = (now - self._win_start_ts) * 1000.0

                # 位图
                S = int(self._slow_slots_per_window)
                bits = self._build_rx_pls_bits(self._win_start_ts, win_ms, S)
                bmap_bytes = bytearray((S + 7) // 8)
                for i, b in enumerate(bits):
                    if b:
                        bmap_bytes[i // 8] |= (1 << (i % 8))

                if self._peer_addr:
                    body = struct.pack(">dffffI",
                                       float(self._win_start_ts),
                                       float(lr), float(la),
                                       float(rx), float(win_ms),
                                       int(self._last_seq_rx or 0))
                    body += struct.pack("B", len(bmap_bytes)) + bytes(bmap_bytes)
                    # NEW: 回显 RTT 请求时间戳（若存在）
                    if self._last_rtcp_req_ts is not None:
                        body += struct.pack("B", 1) + struct.pack(">d", float(self._last_rtcp_req_ts))
                        self._last_rtcp_req_ts = None
                    else:
                        body += struct.pack("B", 0)

                    out = bytes([self.PKT_RTCP_REPORT]) + body
                    try:
                        self._sock.sendto(out, self._peer_addr)
                    except Exception as e:
                        if self._logging:
                            self._LOGGER.debug(f"send RTCP report error: {e}")

                    # 接收端 JSONL
                    # self._write_jsonl_sample(
                    #     win_start_sec=float(self._win_start_ts),
                    #     lr=float(lr),
                    #     la=float(la),
                    #     win_ms=float(win_ms),
                    #     bits_override=bits,
                    #     out_path_override=self._slow_jsonl_path,
                    # )

                    # —— GCC: 发送 REMB(Ar) ——（每次窗口结束按最近状态下发）
                    if self._gcc.enabled:
                        rr = self._gcc_calc_Rr_500ms(now)  # bps
                        ar = self._gcc_update_fsm_and_get_Ar(rr_bps=rr)
                        # print("ar:",ar,"rr:",rr)
                        # print("ar:",ar,"rr:",rr)
                        ar = min(ar, self._gcc.rr_ub_factor * rr)  # 上限约束
                        self._gcc.ar_bps = float(max(1.0, ar))
                        remb = bytes([self.PKT_REMB]) + struct.pack(">d", float(self._gcc.ar_bps))
                        try:
                            self._sock.sendto(remb, self._peer_addr)
                        except Exception as e:
                            if self._logging:
                                self._LOGGER.debug(f"send REMB error: {e}")

                # 窗口复位
                self._win_start_ts = now
                self._win_rx_count = 0
                self._win_expected_count = 0
                self._rx_missing_seq_times.clear()

            # 发送端：周期性滚动落盘
            if not self._recv_mode:
                self._dump_old_tx_records()

            time.sleep(0.01)

    # ===================== 接收端位图构造（原逻辑保持） =====================
    def _build_rx_pls_bits(self, win_start_sec: float, win_ms: float, S: int) -> List[int]:
        S = max(1, int(S))
        bits = [0] * S
        if win_ms <= 0:
            return bits
        slot_ms = float(win_ms) / float(S)
        win_start = float(win_start_sec)
        win_end = win_start + float(win_ms) / 1000.0
        for t in self._rx_missing_seq_times:
            if t < win_start or t >= win_end:
                continue
            rel_ms = (t - win_start) * 1000.0
            idx = int(rel_ms // slot_ms)
            if idx < 0:
                idx = 0
            elif idx >= S:
                idx = S - 1
            bits[idx] = 1
        return bits

    # ===================== JSONL 写入（原逻辑保持） =====================
    def _write_jsonl_sample(self, win_start_sec: float, lr: float, la: float, win_ms: float,
                            bits_override: Optional[List[int]] = None,
                            out_path_override: Optional[str] = None) -> None:
        try:
            S = int(self._slow_slots_per_window)
            B = int(self._slow_bytes_per_window)

            if bits_override is not None:
                bits = list(bits_override)
            else:
                bits = self._hist_ls_bits[-1] if self._hist_ls_bits else [0] * S

            by = bytearray(B)
            for i, bit in enumerate(bits):
                if bit:
                    by[i // 8] |= (1 << (i % 8))

            b64 = base64.b64encode(bytes(by)).decode("ascii")

            rec = {
                "ts_ms": int(float(win_start_sec) * 1000.0),
                "lr": float(lr),
                "la": float(la),
                "pls_bits_b64": b64,
                "window_ms": float(win_ms),
                "pls_slot_ms": float(win_ms) / max(1, S),
            }
            path = out_path_override or self._slow_jsonl_path
            line = json.dumps(rec, ensure_ascii=False) + "\n"
            self._logq.put((path, line))
        except Exception as e:
            if self._logging:
                self._LOGGER.debug(f"_write_jsonl_sample error: {e}")

    # ===================== 异步日志落盘线程（原逻辑保持） =====================
    def _log_sink_loop(self) -> None:
        buffers: Dict[str, List[str]] = {}
        last_flush = time.time()
        FLUSH_LINES = 200
        FLUSH_SEC = 0.2
        while not self._terminate:
            try:
                path, line = self._logq.get(timeout=0.05)
                buf = buffers.setdefault(path, [])
                buf.append(line)
            except Exception:
                pass
            now = time.time()
            if (now - last_flush) >= FLUSH_SEC:
                for path, buf in list(buffers.items()):
                    if not buf:
                        continue
                    try:
                        os.makedirs(os.path.dirname(path), exist_ok=True)
                        with open(path, "a", encoding="utf-8") as f:
                            f.write("".join(buf))
                    except Exception:
                        pass
                    buffers[path].clear()
                last_flush = now
            else:
                for path, buf in list(buffers.items()):
                    if len(buf) >= FLUSH_LINES:
                        try:
                            os.makedirs(os.path.dirname(path), exist_ok=True)
                            with open(path, "a", encoding="utf-8") as f:
                                f.write("".join(buf))
                        except Exception:
                            pass
                        buffers[path].clear()

    # ===================== 慢模块在线前向（原逻辑保持） =====================
    # def _run_slow_forward(self) -> None:
    #     if self._slow_model is None:
    #         return
    #     try:
    #         import torch  # 本地 CPU 前向足够
    #         n = int(self._slow_hist_n)
    #         S = int(self._slow_slots_per_window)

    #         wins = list(self._hist_ls_bits)[-n:]
    #         if len(wins) < n:
    #             pad = [[0]*S for _ in range(n - len(wins))]
    #             wins = pad + wins

    #         concat_bits: List[int] = []
    #         for w in wins:
    #             concat_bits.extend(w)

    #         P_lr = list(self._hist_lr)[-n:]
    #         P_la = list(self._hist_la)[-n:]
    #         if len(P_lr) < n:
    #             P_lr = [0.0]*(n - len(P_lr)) + P_lr
    #         if len(P_la) < n:
    #             P_la = [0.0]*(n - len(P_la)) + P_la

    #         with torch.no_grad():
    #             y = self._slow_model(P_lr=P_lr, P_la=P_la, l_s_bits=concat_bits)  # type: ignore
    #         lr_f = float(y[0].item())
    #         la_f = float(y[1].item())
    #         self._slow_pred = {"lr_f": lr_f, "la_f": la_f}
    #     except Exception as e:
    #         if self._logging:
    #             self._LOGGER.debug(f"_run_slow_forward error: {e}")

    # ===================== 发送端包级 ACK 记录滚动落盘（原逻辑保持） =====================
    def _dump_old_tx_records(self, force: bool = False) -> None:
        try:
            now = time.time()
            if (not force) and ((now - self._last_tx_dump_ts) < 10.0):
                return
            self._last_tx_dump_ts = now
            cutoff_dump = now - 10.0
            cutoff_force = now - 20.0

            out_lines = []
            while self._tx_records:
                rec = self._tx_records[0]
                ts_send = float(rec.get("ts_send", now))
                if ts_send <= cutoff_force:
                    self._tx_records.popleft()
                    self._tx_index.pop(int(rec.get("seq", -1)), None)
                    line = {
                        "seq": int(rec.get("seq", -1)),
                        "ts_send_ms": int(round(ts_send * 1000.0)),
                        "ts_ack_ms": (None if rec.get("ts_ack") is None else int(round(float(rec["ts_ack"]) * 1000.0)))
                    }
                    out_lines.append(json.dumps(line, ensure_ascii=False) + "\n")
                elif ts_send <= cutoff_dump:
                    if rec.get("ts_ack") is not None:
                        self._tx_records.popleft()
                        self._tx_index.pop(int(rec.get("seq", -1)), None)
                        line = {
                            "seq": int(rec.get("seq", -1)),
                            "ts_send_ms": int(round(ts_send * 1000.0)),
                            "ts_ack_ms": int(round(float(rec["ts_ack"]) * 1000.0)),
                        }
                        out_lines.append(json.dumps(line, ensure_ascii=False) + "\n")
                    else:
                        break
                else:
                    break
            for s in out_lines:
                self._logq.put((self._tx_pkt_log_path, s))
        except Exception as e:
            if self._logging:
                self._LOGGER.debug(f"_dump_old_tx_records error: {e}")

    # ===================== NEW: pacer 线程 =====================
    def _pacer_loop(self) -> None:
        """
        在 GCC 模式下，按照 target_bps 的节奏从队列中取包并通过 send() 发出。
        - 队列积压超过阈值时，临时用 pacing_factor 提高瞬时速率，加速清队列。
        - NEW: 若存在 30ms 突发窗口，则窗口内直接发送（不受 bytes_budget 阻挡），但仍统一计费。
        """
        last_sent_ts = time.time()
        bytes_budget = 0.0
        while not self._terminate:
            if not self._gcc.enabled:
                time.sleep(0.001)
                continue

            now = time.time()
            # === 媒体/FEC 队列过期清理 ===
            with self._pacer_lock:
                while self._pacer_queue and self._pacer_queue[0][3] < now:
                    self._pacer_queue.popleft()
                    self._pacer_drop_expired += 1
                while self._pacer_fec_queue and self._pacer_fec_queue[0][3] < now:
                    self._pacer_fec_queue.popleft()
                    self._pacer_fec_drop_expired += 1

            # 预算累积（常规）
            dt = max(0.0, now - last_sent_ts)
            last_sent_ts = now
            A = self.get_gcc_target_bps()
            if not math.isfinite(A) or A <= 0:
                A = 1.0

            effective_A = A
            with self._pacer_lock:
                backlog = len(self._pacer_queue)
            if backlog >= self._gcc.backlog_boost_threshold:
                effective_A = A * self._gcc.pacing_factor
            bytes_budget += (effective_A / 8.0) * dt  # bps -> B/s * dt

            sent_any = False
            while True:
                now2 = time.time()
                # 30ms 窗口是否有效
                burst_active = (now2 < self._burst_until_ts) and (self._burst_bytes_left > 0)

                with self._pacer_lock:
                    # 取包：媒体优先，其次冗余
                    if self._pacer_queue:
                        src_queue = self._pacer_queue
                    elif self._pacer_fec_queue:
                        src_queue = self._pacer_fec_queue
                    else:
                        src_queue = None

                    if src_queue is None:
                        break  # 没有可发的包

                    payload, pkt_type, fid, ddl = src_queue[0]
                    if ddl < now2:
                        # 再次过期检查
                        src_queue.popleft()
                        if src_queue is self._pacer_queue:
                            self._pacer_drop_expired += 1
                        else:
                            self._pacer_fec_drop_expired += 1
                        continue

                    need = len(payload)

                    # 判定是否可发：
                    # - 常规：bytes_budget >= need
                    # - 突发：burst_active → 忽略预算，直接允许；并从 _burst_bytes_left 扣除
                    can_send_normally = (bytes_budget >= need)
                    can_send_burst = burst_active

                    if (not can_send_normally) and (not can_send_burst):
                        break  # 既不够预算，又没有突发窗口 → 暂停

                    # 发送
                    try:
                        self.send(payload, pkt_type=pkt_type)
                    except Exception:
                        src_queue.popleft()
                        continue

                    # 计费：统一扣预算（允许为负）
                    bytes_budget -= need

                    # 若突发路径，扣突发额度（不阻挡）
                    if can_send_burst:
                        self._burst_bytes_left = max(0, self._burst_bytes_left - need)

                    # 出队
                    src_queue.popleft()
                    sent_any = True

                # 节流睡眠（避免一次性倾倒过多 syscalls）
                time.sleep(self._gcc.pacer_sleep_ns / 1e9)

                # 突发窗口结束条件（时间到或额度用尽）
                if (time.time() >= self._burst_until_ts) or (self._burst_bytes_left <= 0):
                    self._burst_until_ts = 0.0
                    self._burst_bytes_left = 0

                # 若不是突发，且预算已不足，结束内层循环
                if (not burst_active) and (bytes_budget <= 0):
                    break

            if not sent_any:
                time.sleep(0.0005)

    # ===================== NEW: 接收端 GCC arrival 解析/状态推进 =====================
    def _gcc_got_gcc_data(self, frame_id: int, Ti: float, is_first: bool, is_last: bool, t_arrival: float) -> None:
        """
        仅在接收端、GCC 启用且数据帧携带 GCC 头时调用。
        - 构建“帧”的 (Ti, ti)；在 LAST 到达时计算 dm = (ti-ti_prev) - (Ti-Ti_prev) 并进入滤波/阈值/状态机。
        - 吞吐统计在 _io_loop() 中以字节数累计，这里只做时序与 dm。
        """
        # 帧边界管理
        if is_first:
            self._gcc.cur_frame_id = frame_id
            self._gcc.cur_Ti = float(Ti)
            self._gcc.cur_first_seen = float(t_arrival)
            self._gcc.cur_last_seen = float(t_arrival)
            self._gcc.cur_received_any = True
        else:
            # 中间包更新末时间
            if self._gcc.cur_received_any:
                self._gcc.cur_last_seen = float(t_arrival)

        if is_last and self._gcc.cur_received_any:
            Ti_cur = float(self._gcc.cur_Ti if self._gcc.cur_Ti is not None else Ti)
            ti_cur = float(self._gcc.cur_last_seen if self._gcc.cur_last_seen is not None else t_arrival)

            if (self._gcc.prev_Ti is not None) and (self._gcc.prev_ti is not None):
                dm = (ti_cur - float(self._gcc.prev_ti)) - (Ti_cur - float(self._gcc.prev_Ti))
                self._gcc_kalman_like_update(dm, ti_cur)

            self._gcc.prev_Ti = Ti_cur
            self._gcc.prev_ti = ti_cur
            # 重置当前帧跟踪
            self._gcc.cur_frame_id = None
            self._gcc.cur_Ti = None
            self._gcc.cur_first_seen = None
            self._gcc.cur_last_seen = None
            self._gcc.cur_received_any = False

    def _gcc_kalman_like_update(self, dm: float, ti: float) -> None:
        """
        Kalman 简化：m(t) = (1-K)m(t-1) + K*dm；K ~ Q/(Q + sigma2)，sigma2 用残差 EWMA 估计。
        同时更新自适应阈值 gamma，并据 m/gamma 触发 overuse/underuse/normal。
        """
        # 残差方差估计
        z = float(dm - self._gcc.m_t)
        self._gcc.resid_var = float(self._gcc.beta_noise * self._gcc.resid_var + (1.0 - self._gcc.beta_noise) * (z * z))
        sigma2 = max(1e-9, self._gcc.resid_var)

        # Kalman 增益近似
        K = float(self._gcc.Q / (self._gcc.Q + sigma2))
        self._gcc.m_t = float((1.0 - K) * self._gcc.m_t + K * dm)

        # 自适应阈值 gamma(t)
        delta_T = 1.0 / 30.0
        k = self._gcc.kd if (abs(self._gcc.m_t) < self._gcc.gamma) else self._gcc.ku
        self._gcc.gamma = float(self._gcc.gamma + delta_T * k * (abs(self._gcc.m_t) - self._gcc.gamma))

        # 过载/欠载/正常
        over = (self._gcc.m_t > self._gcc.gamma)
        under = (self._gcc.m_t < -self._gcc.gamma)
        if over:
            self._gcc.state = "Decrease"
        elif under:
            self._gcc.state = "Hold"
        else:
            self._gcc.state = "Increase"

    # ===================== NEW: 500ms Rr 估计 =====================
    def _gcc_calc_Rr_500ms(self, now_sec: float) -> float:
        WIN = 0.5
        q = self._gcc.recv_bytes_500ms
        while q and (now_sec - q[0][0] > WIN):
            q.popleft()
        total_bytes = sum(b for _, b in q)
        rr_bps = (total_bytes * 8.0) / WIN if WIN > 0 else 0.0
        return float(max(1e5, rr_bps))

    # ===================== NEW: FSM → Ar 输出 =====================
    def _gcc_update_fsm_and_get_Ar(self, rr_bps: float) -> float:
        Ar_prev = float(self._gcc.ar_bps if math.isfinite(self._gcc.ar_bps) else rr_bps)
        if self._gcc.state == "Increase":
            Ar = max(1.0, self._gcc.eta_inc * Ar_prev)
        elif self._gcc.state == "Decrease":
            Ar = max(1.0, self._gcc.alpha_dec * rr_bps)
        else:
            Ar = Ar_prev
        return float(Ar)

    # ===================== NEW: A 融合与 EMA 平滑 =====================
    def _fuse_A_and_apply_ema(self) -> None:
        A_raw = float(min(self._gcc.ar_bps, self._gcc.as_bps))
        if A_raw <= 0 or not math.isfinite(A_raw):
            A_raw = 1.0
        self._gcc.target_bps = A_raw
        self._apply_A_ema()

    def _apply_A_ema(self) -> None:
        A = float(self._gcc.target_bps if math.isfinite(self._gcc.target_bps) else 1.0)
        if not math.isfinite(self._gcc.target_bps_ema):
            self._gcc.target_bps_ema = A
        else:
            alpha = min(max(self._A_ema_alpha, 0.0), 1.0)
            self._gcc.target_bps_ema = alpha * A + (1.0 - alpha) * self._gcc.target_bps_ema

    # ===================== NEW: 基于最近 lr 的 As 止损 =====================
    def _update_As_from_loss(self, lr: float) -> None:
        if lr >= self._loss_lr_thresh and math.isfinite(self._gcc.as_bps):
            self._gcc.as_bps = max(1.0, self._gcc.as_bps * self._loss_as_decay)
            self._fuse_A_and_apply_ema()

    def get_pacer_drops(self) -> Dict[str, int]:
        return {
            "expired": int(self._pacer_drop_expired),
            "overflow": int(self._pacer_drop_overflow),
            "queue_len": int(len(self._pacer_queue)),
            "queue_cap": int(self._pacer_max_pkts),
        }
